package com.edu.yx.app.App_04_DwsApp.Dws_02_User;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.edu.yx.app.APP_01_BaseApp.BaseApp_moreTopic;
import com.edu.yx.bean.UserFunnelBehaviorBean;
import com.edu.yx.util.TempUtil;
import com.edu.yx.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;

import static com.edu.yx.common.Constant.*;

public class Dws_02_User_Funnel_behavior extends BaseApp_moreTopic {

    public static void main(String[] args) {
        new Dws_02_User_Funnel_behavior().init(
                4022,
                2,
                "Dws_02_User_Funnel_behavior",
                TOPIC_DWD_TRAFFIC_PAGE, //页面行为首页浏览、商品浏览
                TOPIC_DWD_TRADE_CART_ADD, //加购
                TOPIC_DWD_TRADE_ORDER_INFO, //下单人数
                TOPIC_DWD_TRADE_PAY_DETAIL_SUC //成功支付
        );
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> streams) {

        //1.封装成bean 并且union到一起
        DataStream<UserFunnelBehaviorBean> unionAll = unionAll(streams);

        //2.开窗聚合
        SingleOutputStreamOperator<UserFunnelBehaviorBean> windowAndAgg = windowAndAgg(unionAll);

        //3.写入doris
        writeToDoris(windowAndAgg);

    }

    private void writeToDoris(SingleOutputStreamOperator<UserFunnelBehaviorBean> windowAndAgg) {

        windowAndAgg
                .map(bean ->{
                    //配置json可以将驼峰转换成带下划线，然后方便导入数据到doris
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;

                    String json = JSON.toJSONString(bean, config);
//                    System.out.println(json);
                    return json;
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_user_funnel_behavior_count"));
    }


    private SingleOutputStreamOperator<UserFunnelBehaviorBean> windowAndAgg(DataStream<UserFunnelBehaviorBean> unionAll) {

        SingleOutputStreamOperator<UserFunnelBehaviorBean> result = unionAll
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<UserFunnelBehaviorBean>forBoundedOutOfOrderness(Duration.ofSeconds(15))
                        .withTimestampAssigner((bean, ts) -> bean.getTs())
                )
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sideOutputLateData(new OutputTag<UserFunnelBehaviorBean>("late") {
                })
                .reduce(new ReduceFunction<UserFunnelBehaviorBean>() {
                            @Override
                            public UserFunnelBehaviorBean reduce(UserFunnelBehaviorBean value1,
                                                                 UserFunnelBehaviorBean value2) throws Exception {

                                //处理普通加法
                                value1.setHomeUvCt(value1.getHomeUvCt() + value2.getHomeUvCt());
                                value1.setCourseDetailUvCt(value1.getCourseDetailUvCt() + value2.getCourseDetailUvCt());

                                //处理集合
                                value1.getOrderUserSet().addAll(value2.getOrderUserSet());

                                value1.getCarAddUserSet().addAll(value2.getCarAddUserSet());
                                value1.getOrderPaySucUserSet().addAll(value2.getOrderPaySucUserSet());

                                return value1;
                            }
                        }, new ProcessAllWindowFunction<UserFunnelBehaviorBean, UserFunnelBehaviorBean, TimeWindow>() {
                            @Override
                            public void process(Context context,
                                                Iterable<UserFunnelBehaviorBean> elements,
                                                Collector<UserFunnelBehaviorBean> out) throws Exception {

                                UserFunnelBehaviorBean bean = elements.iterator().next();
                                bean.setStt(TempUtil.toDateTime(context.window().getStart()));
                                bean.setEdt(TempUtil.toDateTime(context.window().getEnd()));
                                bean.setCurDate(TempUtil.toDate(System.currentTimeMillis()));

                                //去除集合内的null
                                bean.getOrderPaySucUserSet().removeAll(Collections.singleton("null"));
                                bean.getOrderUserSet().removeAll(Collections.singleton("null"));
                                bean.getCarAddUserSet().removeAll(Collections.singleton("null"));

                                bean.setCarAddUserCt((long) bean.getCarAddUserSet().size());
                                bean.setOrderUserCt((long) bean.getOrderUserSet().size());
                                bean.setOrderPaySucUserCt((long) bean.getOrderPaySucUserSet().size());

                                out.collect(bean);

                            }
                        }

                );

//        result.print("main");
//        result.getSideOutput(new OutputTag<UserFunnelBehaviorBean>("late"){}).print("late");
        return result;
    }

    private DataStream<UserFunnelBehaviorBean> unionAll(HashMap<String, DataStreamSource<String>> streams) {
        //1.首页浏览人数
        SingleOutputStreamOperator<UserFunnelBehaviorBean> homeAndCourseDetailStream = streams
                //todo 这个流负责读取首页浏览人数
                .get(TOPIC_DWD_TRAFFIC_PAGE)
                .filter(strjson-> {
                    JSONObject obj = JSONObject.parseObject(strjson);
                    //todo 从大的json对象内获取小的对象
//                    JSONObject common = obj.getJSONObject("common");
                    JSONObject page = obj.getJSONObject("page");

                    return page.getString("page_id").equals("home") || page.getString("page_id").equals("course_detail");
                })
                .map(strjson -> {

                    //todo 将kafka拿到的json格式的字符串转换成json对象
                    JSONObject obj = JSONObject.parseObject(strjson);

                    //todo 从大的json对象内获取小的对象
//                    JSONObject common = obj.getJSONObject("common");
                    JSONObject page = obj.getJSONObject("page");

                    //todo 首页浏览数
                    Long homeUvCt = page.getString("page_id").equals("home") ? 1L: 0L ;
                    Long courserDetailUvCt = page.getString("page_id").equals("course_detail") ? 1L: 0L ;

                    //todo 用于提供进行窗口聚合的事件时间，单位为毫秒，不用乘以1000L，
                    //由于page的数据内时间是紊乱的，这里采用系统的时间 由于这个是kafka的数据直接拿过来，让他慢5秒
//                    Long ts = System.currentTimeMillis() + 10000;
//                    System.out.println(ts);
                    Long ts = obj.getLong("ts");

                    return new UserFunnelBehaviorBean(
                            "", //下游窗口计算生成
                            "",
                            "",
                            homeUvCt,
                            courserDetailUvCt,
                            0L,
                            new HashSet<>(Collections.singleton("null")),
                            0L,
                            new HashSet<>(Collections.singleton("null")),
                            0L,
                            new HashSet<>(Collections.singleton("null")),
                            ts
                    );
                });

        //2.商品详情页浏览数
//        SingleOutputStreamOperator<UserFunnelBehaviorBean> courseDetailStream = streams
//                //todo 这个流负责读取首页浏览人数
//                .get(TOPIC_DWD_TRAFFIC_PAGE)
//                .map(strjson -> {
//
//                    //todo 将kafka拿到的json格式的字符串转换成json对象
//                    JSONObject obj = JSONObject.parseObject(strjson);
//
//                    //todo 从大的json对象内获取小的对象
//                    JSONObject common = obj.getJSONObject("common");
//                    JSONObject page = obj.getJSONObject("page");
//
//                    //todo 首页浏览数
//
//
//                    //todo 用于提供进行窗口聚合的事件时间，单位为毫秒，不用乘以1000L
//                    Long ts = obj.getLong("ts");
//
//                    return new UserFunnelBehaviorBean(
//                            "", //下游窗口计算生成
//                            "",
//                            "",
//                            0L, //下游计算生成
//                            courserDetailUvCt,
//                            0L,
//                            new HashSet<>(Collections.singleton("null")),
//                            0L,
//                            new HashSet<>(Collections.singleton("null")),
//                            0L,
//                            new HashSet<>(Collections.singleton("null")),
//                            ts
//                    );
//                });


        //3.加购人数
        SingleOutputStreamOperator<UserFunnelBehaviorBean> cartUserStream = streams
                //todo 这个流负责读取首页浏览人数
                .get(TOPIC_DWD_TRADE_CART_ADD)
                .map(strjson -> {

                    //todo 将kafka拿到的json格式的字符串转换成json对象
                    JSONObject obj = JSONObject.parseObject(strjson);

                    String user_id = obj.getString("user_id");


                    //todo 用于提供进行窗口聚合的事件时间，单位为秒，乘以1000L
                    Long ts = obj.getLong("ts") * 1000L;
//                    Long ts = System.currentTimeMillis() - 3000L;
                    return new UserFunnelBehaviorBean(
                            "", //下游窗口计算生成
                            "",
                            "",
                            0L, //下游计算生成
                            0L,
                            0L,
                            new HashSet<>(Collections.singleton(user_id)),
                            0L,
                            new HashSet<>(Collections.singleton("null")),
                            0L,
                            new HashSet<>(Collections.singleton("null")),
                            ts
                    );
                });

        //4.下单人数
        SingleOutputStreamOperator<UserFunnelBehaviorBean> orderUserStream = streams
                //todo 这个流负责读取首页浏览人数
                .get(TOPIC_DWD_TRADE_ORDER_INFO)
                .map(strjson -> {

                    //todo 将kafka拿到的json格式的字符串转换成json对象
                    JSONObject obj = JSONObject.parseObject(strjson);

                    String user_id = obj.getString("user_id");


                    //todo 用于提供进行窗口聚合的事件时间，单位为秒，乘以1000L
                    Long ts = obj.getLong("ts") * 1000L;
//                    Long ts = System.currentTimeMillis() - 3000L;

                    return new UserFunnelBehaviorBean(
                            "", //下游窗口计算生成
                            "",
                            "",
                            0L, //下游计算生成
                            0L,
                            0L,
                            new HashSet<>(Collections.singleton("null")),
                            0L,
                            new HashSet<>(Collections.singleton(user_id)),
                            0L,
                            new HashSet<>(Collections.singleton("null")),
                            ts
                    );
                });

        //5.下单成功人数
        SingleOutputStreamOperator<UserFunnelBehaviorBean> orderPaySucUserStream = streams
                //todo 这个流负责读取首页浏览人数
                .get(TOPIC_DWD_TRADE_PAY_DETAIL_SUC)
                .map(strjson -> {

                    //todo 将kafka拿到的json格式的字符串转换成json对象
                    JSONObject obj = JSONObject.parseObject(strjson);

                    String user_id = obj.getString("user_id");

                    //todo 用于提供进行窗口聚合的事件时间，单位为秒，乘以1000L
                    Long ts = obj.getLong("ts") * 1000L;
//                    Long ts = System.currentTimeMillis() - 3000L;
                    return new UserFunnelBehaviorBean(
                            "", //下游窗口计算生成
                            "",
                            "",
                            0L, //下游计算生成
                            0L,
                            0L,
                            new HashSet<>(Collections.singleton("null")),
                            0L,
                            new HashSet<>(Collections.singleton("null")),
                            0L,
                            new HashSet<>(Collections.singleton(user_id)),
                            ts
                    );
                });



        return homeAndCourseDetailStream.union(cartUserStream,orderUserStream,orderPaySucUserStream);
    }

}
